import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;


public class KafkaCon {


    public static void main(String[] args) {
        // 1) 配置项  kafka 的服务地址
        // 2) 配置   k v String
        //3) 创建 kafka 的 消费者组 ，并设置kafka的消费者ID
        //4) 订阅 kafka的 topic
        //5) 通过循环 consumerRecorde 拉取 主题当中的数据

        Properties proper = new Properties();
        proper.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"123.56.187.176:9092");
        proper.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG ,"org.apache.kafka.common.serialization.StringDeserializer");
        proper.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG ,"org.apache.kafka.common.serialization.StringDeserializer");
        proper.put(ConsumerConfig.GROUP_ID_CONFIG,"niit");

        proper.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String, String>(proper);

        //topic name
        kafkaConsumer.subscribe(Arrays.asList("stuInfo"));


        // 循环 消费 topic 当中的数据
        while (true){

            ConsumerRecords<String, String> conRecord = kafkaConsumer.poll(5);

            for(ConsumerRecord<String,String> record :conRecord){
//                System.out.println("value is " + record);
                System.out.println("value is " + record.value());
            }



        }

    }


}